package com.alibaba.bink;

import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.util.stream.Stream;

/**
 * @author galeno
 * @Title:
 * @Description:
 * @date 2021/10/2815:46
 * /**
 *  * 两个数据流，使用EventTime划分窗口，实现LeftOuterJoin
 *  *
 *  * 窗口触发的条件为，每一个流，每一个分区都有满足窗口触发的条件
 *  *  - 每个分区的WaterMark为该分区中最大的EventTime - 延迟时间
 *  *  - 窗口内所有subtask的waterMark为上游生成WaterMarter所有分区中最下的作为每个分区即该窗口的WaterMark
 *  */
public class LeftJoinDemo {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port",12345);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        //左边的流
        DataStreamSource<String> leftLines = env.socketTextStream("galeno", 9998);
        //1000,c10,2000
        SingleOutputStreamOperator<Tuple3<Long, String, Integer>> leftStream = leftLines.map(new MapFunction<String, Tuple3<Long, String, Integer>>() {
            @Override
            public Tuple3<Long, String, Integer> map(String value) throws Exception {
                String[] split = value.split(",");
                return Tuple3.of(Long.parseLong(split[0]), split[1], Integer.parseInt(split[2]));

            }
        }).setParallelism(2);
        SingleOutputStreamOperator<Tuple3<Long, String, Integer>> leftStreamWaterMark = leftStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Tuple3<Long, String, Integer>>(Time.seconds(0)) {
            @Override
            public long extractTimestamp(Tuple3<Long, String, Integer> element) {
                return element.f0;
            }
        });
        //1000,c10,图书
        DataStreamSource<String> lines2 = env.socketTextStream("galeno", 9999);
        SingleOutputStreamOperator<Tuple3<Long, String, String>> rightStream = lines2.map(new MapFunction<String, Tuple3<Long, String, String>>() {
            @Override
            public Tuple3<Long, String, String> map(String value) throws Exception {
                String[] split = value.split(",");
                return Tuple3.of(Long.parseLong(split[0]), split[1], split[2]);

            }
        }).setParallelism(2);
        SingleOutputStreamOperator<Tuple3<Long, String, String>> rightWaterMark = rightStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Tuple3<Long, String, String>>(Time.seconds(0)) {
            @Override
            public long extractTimestamp(Tuple3<Long, String, String> element) {
                return element.f0;
            }
        });
        DataStream<Tuple5<Long, String, Integer, Long, String>> res = leftStreamWaterMark.coGroup(rightWaterMark)
                .where(x -> x.f1)
                .equalTo(x -> x.f1)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .apply(new CoGroupFunction<Tuple3<Long, String, Integer>, Tuple3<Long, String, String>, Tuple5<Long, String, Integer, Long, String>>() {
                    @Override
                    public void coGroup(Iterable<Tuple3<Long, String, Integer>> first, Iterable<Tuple3<Long, String, String>> second, Collector<Tuple5<Long, String, Integer, Long, String>> out) throws Exception {
                        for (Tuple3<Long, String, Integer> left : first) {
                            boolean isJoined = false;
                            for (Tuple3<Long, String, String> right : second) {
                                isJoined = true;
                                out.collect(Tuple5.of(left.f0, left.f1, left.f2, right.f0, right.f2));
                            }
                            if (!isJoined) {
                                out.collect(Tuple5.of(left.f0, left.f1, left.f2, null, null));
                            }
                        }
                    }
                });
        res.writeAsText("D://a.txt", FileSystem.WriteMode.OVERWRITE).setParallelism(1);
        res.print();
        env.getExecutionPlan();
        env.execute();


    }
}
